/*
 *    Copyright 2008 Tim Jansen
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package org.actorsguildframework.internal;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;

import org.actorsguildframework.Actor;
import org.actorsguildframework.DefaultAgent;
import org.actorsguildframework.ActorException;
import org.actorsguildframework.AsyncResult;
import org.actorsguildframework.ActorRuntimeException;
import org.actorsguildframework.annotations.ConcurrencyModel;
import org.actorsguildframework.annotations.ThreadUsage;
import org.actorsguildframework.internal.util.FastQueue;
import org.actorsguildframework.internal.util.L;


/**
 * Describes the state of an actor.
 */
public final class ActorState extends FastQueue.Entry{
	private final static L log = new L(DefaultAgent.class);
	private final static AtomicLong actorIdGenerator = new AtomicLong();
	
	/**
	 * A reference to the Actor's scheduler.
	 */
	private final Controller controller;

	/**
	 * The actor implementation instance the state belongs to.
	 */
	private final Actor actor;
	
	/**
	 * The unique number of the actor (only for logging).
	 */
	private final long actorNumber = actorIdGenerator.addAndGet(1);
	
	/**
	 * Contains the list of all incoming messages that need to be processed in this 
	 * actor and can be processed at any time (MT=Multi-Threaded).
	 * 
	 * Locking policy: you must lock {@link ControllerImplementation#actorLock} before accessing this field.
	 * @see #mailboxST
	 */	
	private final FastQueue<MessageInvocation<?>> mailboxMT;

	/**
	 * Contains the list of all incoming messages that need to be processed in this 
	 * actor and can only be processed when the actor is not busy (ST=Single-Threaded).
	 * 
	 * If a thread processes one of the messages here, it should also process
	 * all other messages until the queue is empty.
	 * 
	 * Locking policy: you must lock {@link ControllerImplementation#actorLock} before accessing this field.
	 * @see #mailboxMT 
	 */	
	private final FastQueue<MessageInvocation<?>> mailboxST;

	
	/**
	 * mIsInitialized is false after the invocation of the actor's constructor, unless the
	 * Actor does not have any initializer - in that case it is true.
	 * After the completion of the initializer mIsInitialized will be set to true.
	 * 
	 * Any attempt to invoke a message while mIsInitialized is false will cause a 
	 * NotInitializedYetException. Any attempt to invoke an initializer when mIsInitialized
	 * is true will cause a AlreadyInitializedException.
	 * 
	 * Locking policy: you must lock {@link ControllerImplementation#actorLock} before accessing this field.
	 */
	private boolean isInitialized;
	
	/**
	 * mIsBusyWith is used for single-threaded actors:
	 * If the actor is currently executing a message, it refers to this message. 
	 * No other thread can access the actor during that time. Otherwise it is null. 
	 * This means that a Thread can invoke a message, provided that 
	 * it can set the mIsBusyWith to itself successfully.
	 *  
	 * Multi-threaded actors don't use this field at all.
	 * 
	 * Locking policy: you must lock {@link ControllerImplementation#actorLock} before accessing this field.
	 */
	private MessageInvocation<?> isBusyWith;

	/**
	 * mNumWaitingThreads contains the number of threads that are currently waiting for the
	 * busy lock ({@link #isBusyWith}). If you are waiting for the busy lock, you should increase
	 * this value. If you got the lock, release it. 
	 * 
	 * If mNumWaitingThreads is >0, the actor will not be in the controller's queue.
	 *
	 * Locking policy: you must lock {@link ControllerImplementation#actorLock} before accessing this field.
	 */
	private int numWaitingThreads;

	/**
	 * This condition of {@link ControllerImplementation#actorLock} will be signaled every time the 
	 * {@link #isBusyWith} field is freed (set to null). This allows threads to wait until
	 * they can access the actor. 
	 */
	private Condition busyLockFreed;
	
	/**
	 * Contains the number of additional threads that could now start running for this actor.
	 * (This is the sum of unprocessed multi-threaded messages plus one of there is a single-threaded
	 * message that could be executed now)  
	 */
	private int numberOfThreadsNeeded;
	
	/**
	 * Creates a new ActorState instance.
	 * @param scheduler the Actor's scheduler
	 * @param actor the actor whose state this instance is representing
	 * @param isInitialized true if the actor is already initialized
	 */
	public ActorState(Controller scheduler, Actor actor, boolean isInitialized) {
		this.controller = scheduler;
		this.actor = actor;
		this.mailboxMT = new FastQueue<MessageInvocation<?>>();
		this.mailboxST = new FastQueue<MessageInvocation<?>>();
		
		this.isInitialized = isInitialized;
		this.busyLockFreed = controller.getActorLock().newCondition();
	}
	
	/**
	 * Returns the initialization state of the actor.
	 * @return true if the actor is initialized, false otherwise
	 */
	public boolean isInitialized() {
		controller.getActorLock().lock();
		try {
			return isInitializedUnsynchronized();
		}
		finally {
			controller.getActorLock().unlock();
		}
	}

	/**
	 * Returns the initialization state of the actor.
	 * When calling this, you must lock {@link ControllerImplementation#actorLock}.
	 * @return true if the actor is initialized, false otherwise
	 */
	public boolean isInitializedUnsynchronized() {
		return isInitialized;
	}
	
	/**
	 * Sets the actor to initialized.
	 * @throws ActorRuntimeException if the Actor is already initialized
	 */
	public void setInitialized() {
		controller.getActorLock().lock();
		try {
			if (isInitialized)
				throw new ActorRuntimeException("The Actor is already initialized. You can invoke the @Initializer method only once.");
			isInitialized = true;
		}
		finally {
			controller.getActorLock().unlock();
		}
	}

	/**
	 * Returns the actor (without proxy).
	 * @return the actual actor
	 */
	public Actor getActor() {
		return actor;
	}

	/**
	 * Adds the given initializer invocation to the single-threaded queue.
	 * 
	 * @param invk the invocation of the initializer
	 */
	private void addInvokationInitializer(MessageInvocation<?> invk) {
		controller.getActorLock().lock();
		try {
			if (isInitializedUnsynchronized())
				throw new ActorRuntimeException("The Actor is already initialized. You can invoke the actor's initializer only once.");
			mailboxST.add(invk);
			updateControllerQueue();
		}
		finally {
			controller.getActorLock().unlock();
		}
	}
	
	/**
	 * Queues an initializer for execution.
	 * @param caller the caller to use
	 * @param usage the thread usage of the message
	 * @param args the arguments of the initializer. Will be copied (deep copy for mutables)
	 * @return the result handle
	 */
	@SuppressWarnings("unchecked")
	public AsyncResultImpl<?> queueInitializer(MessageCaller caller, ThreadUsage usage, Object[] args) {
	
		MessageInvocation superInvk = ThreadState.get().getCurrentInvocation();
		MessageInvocation invk = new MessageInvocation(this, caller, args, superInvk, 
					ConcurrencyModel.SingleThreaded, usage, true);

		if (controller.isLoggingActions())
			log.info("Queueing initializer message #%d for actor #%d (%s): %s(%s)",
					invk.getMessageNumber(),
					ActorState.getState(actor).getActorNumber(), actor.getClass().getName(),
					caller.getMessageName(), Arrays.deepToString(args));

		addInvokationInitializer(invk);
		AsyncResultImpl<?> r = invk.getAsyncResult();
		r.addNotifier(new AsyncResult.Notifier() {
			
			public void resultReady(AsyncResult result) {
				setInitialized();
			}
		});
		return r;
	}
	
	/**
	 * Adds the given message invocation to the appropriate queue.
	 * 
	 * @param invk the invocation of the initializer
	 */
	private void addInvokationMessage(MessageInvocation<?> invk) {
		controller.getActorLock().lock();
		try {
			if (!isInitializedUnsynchronized())
				throw new ActorRuntimeException("The Actor is not initialized yet. You must call one of the methods marked @Initializer before you can use it.");
			if (invk.getConcurrencyModel().isMultiThreadingCapable())
				mailboxMT.add(invk);
			else
				mailboxST.add(invk);
			updateControllerQueue();
		}
		finally {
			controller.getActorLock().unlock();
		}
	}
	
	/**
	 * Queues a message for execution.
	 * @param caller the caller to use
	 * @param model the model of the message 
	 * @param usage the thread usage of the message
	 * @param args the arguments of the message. Will be copied (deep copy for mutables)
	 * @return the result handle
	 */
	@SuppressWarnings("unchecked")
	public AsyncResultImpl<?> queueMessage(MessageCaller caller, ConcurrencyModel model, 
			ThreadUsage usage, Object[] args) {
		
		MessageInvocation superInvk = ThreadState.get().getCurrentInvocation();
		MessageInvocation invk = new MessageInvocation(this, caller, args, superInvk, 
				model, usage, false);

		if (controller.isLoggingActions())
			log.info("Queueing message #%d for #%d (%s): %s(%s)",
					invk.getMessageNumber(),
					ActorState.getState(actor).getActorNumber(), actor.getClass().getName(),
					caller.getMessageName(), Arrays.deepToString(args));

		addInvokationMessage(invk);
		return invk.getAsyncResult();
	}
	
	/**
	 * Executes a message in the current thread. This is used for {@link Actor#lock(Runnable)}.
	 * @param ts the current ThreadState
	 * @param caller the caller to use
	 * @return the result of the message
	 */
	@SuppressWarnings("unchecked")
	public AsyncResult<?> executeMessageInThread(ThreadState ts, MessageCaller caller) {
		MessageInvocation superInvk = ts.getCurrentInvocation();
		MessageInvocation invk = new MessageInvocation(this, caller, null, superInvk, 
				ConcurrencyModel.SingleThreaded, ThreadUsage.CpuBound, false);
		invk.executeSingleThreadedNow(ts);
		return invk.getAsyncResult();
	}
	
	/**
	 * Returns the number of threads this actor has for for that can be immediately executed.
	 * @return the number of threads that the Actor has work for
	 */
	public int hasWorkForUnsynchronized() {
		return (((isBusyWith == null) && (numWaitingThreads == 0)&& !mailboxST.isEmpty()) ? 1 : 0) +
		 mailboxMT.size();
	}
	
	/**
	 * Removes the given multi-threaded message from the queue.
	 * @param message the message to remove. Must be multi-threaded
	 * @return true if the message has been removed. False if it was not in the queue anymore.
	 */
	public boolean removeMultiThreadedMessageFromQueue(MessageInvocation<?> message) {
		assert message.getConcurrencyModel().isMultiThreadingCapable();
		boolean success;
		controller.getActorLock().lock();
		try {	
			if (!message.getTargetActor().isInitializedUnsynchronized())
				return false; // no MT messages for uninitialized actors!!
			
			success = mailboxMT.remove(message);
			updateControllerQueue();	
		}
		finally {
			controller.getActorLock().unlock();
		}
		return success;
	}
	
	/**
	 * Acquires the busy lock of the Actor. The message must not be queued 
	 * when calling this. Blocks until busy lock is available. 
	 * @param ts the current ThreadState
	 * @param message the message that wants to re-acquire the lock
	 */
	public void acquireBusyLock(ThreadState ts, MessageInvocation<?> message) {
		controller.getActorLock().lock();
		try {
			if (isBusyWith != null) {
				WorkerState oldState = ts.setState(WorkerState.WaitingInternal);
				numWaitingThreads++;					
				updateControllerQueue();	
				try {
					while (isBusyWith != null) 
						busyLockFreed.await();
				}
				catch (InterruptedException e) {
					throw new ActorException("Got interrupted while waiting");
				}
				finally {
					numWaitingThreads--;
				}
				ts.setState(oldState);
			}
		
			isBusyWith = message;
			updateControllerQueue();	
		}
		finally {
			controller.getActorLock().unlock();
		}
	}
	
	/**
	 * Frees the busy lock of the Actor. Adds the Actor into the Controller's queue
	 * if there are other messages.
	 * @param message the message that wants to release the lock
	 */
	public void freeBusyLockAndUpdateQueues(MessageInvocation<?> message) {
		controller.getActorLock().lock();
		try {
			assert isBusyWith == message;
			isBusyWith = null;
			busyLockFreed.signal();

			updateControllerQueue();
		}
		finally {
			controller.getActorLock().unlock();
		}
	}
	

	/**
	 * Finds the next message to execute. Returns it, or null if no message
	 * can be executed.
	 * You must synchronize on this instance before calling!
	 * @return the message, or null
	 */
	private MessageInvocation<?> getMessageToExecUnsynchronized() {
		int mailboxStSize = mailboxST.size();
		int mailboxMtSize = mailboxMT.size();
		if (mailboxStSize+mailboxMtSize == 0)
				return null; // nothing to do -> leave
		
		if ((isBusyWith == null) && (numWaitingThreads == 0)) {
			// not busy: decide whether to take an ST or an MT message
			if (!isInitialized) {
				if (mailboxStSize == 0)
					return null; // no initializer yet -> leave
				
				if (mailboxST.peek().isInitializer())
					return mailboxST.pop();
				else {
					for (MessageInvocation<?> invc: mailboxST)
						if (invc.isInitializer()) {
							mailboxST.remove(invc);
							return invc;
						}
					return null; // no initializer found -> very strange, just leave
				}
			}
			else if (mailboxStSize > 0) { // take the ST msg if lock is free!
				return mailboxST.pop();
			}
			else {
				return mailboxMT.pop();
			}
		}
		else { // busy: we need to take MT
			if (mailboxMtSize == 0)
				return null; // busy and not MT messages -> leave
			return mailboxMT.pop();
		}
	}
	
	/**
	 * Uses the invoking thread to execute all messages in the Actor's queues. May
	 * run for a long time or even forever, especially if the messages send more messages
	 * to the Actor.
	 * You must lock the {@link ControllerImplementation#actorLock} before calling this method.
	 * @param ts the current ThreadState
	 * @param keepRunning a interface that can be used to stop the operation: if it is set, the 
	 *                    method will return as soon as it processed the current message
	 * @return the number of messages that have been executed
	 */
	public int executeAllQueuedMessagesUnsynchronized(ThreadState ts, KeepRunningInterface keepRunning) {
		int msgsExecuted = 0;
		while (keepRunning.shouldContinue()) {
			MessageInvocation<?> msg = getMessageToExecUnsynchronized();
			if (msg == null) {// no msg left -> leave
				updateControllerQueue();
				return msgsExecuted;
			}
			
			boolean isMt = msg.getConcurrencyModel().isMultiThreadingCapable();
			if (!isMt)
				isBusyWith = msg;
			updateControllerQueue();
			controller.getActorLock().unlock(); // unlock for the execution!!
			try {
				msg.runMessageNowNoLocking(ts);
			}
			finally {
				controller.getActorLock().lock();
			}

			if (!isMt) {
				isBusyWith = null;
				busyLockFreed.signal(); 
			}
			msgsExecuted++;
		}
		return msgsExecuted;
	}

	/**
	 * Tries to execute all single-threaded messages in the queue up until the given
	 * final message, but only if the actor is not already busy. If the message is
	 * not queued anymore, the method returns immediately.
	 * @param ts the current ThreadState
	 * @param finalMsg the last message to execute. Must be single-threaded
	 * @return true if the finalMsg is still in the queue, false otherwise
	 */
	public boolean tryExecuteAllSTMessagesUntil(ThreadState ts, MessageInvocation<?> finalMsg) {
		assert !finalMsg.getConcurrencyModel().isMultiThreadingCapable();
		controller.getActorLock().lock();
		try {
			if ((isBusyWith != null) || !mailboxST.isInQueue(finalMsg))
				return false; // quick return
			
			boolean success = false;
			do  {
				MessageInvocation<?> msg = mailboxST.pop();
				if (msg == null) // no message left -> leave (shouldn't happen)
					break;
				
				isBusyWith = msg;
				updateControllerQueue();
				controller.getActorLock().unlock(); // unlock for the execution!!
				try {
					msg.runMessageNowNoLocking(ts);
				}
				finally {
					controller.getActorLock().lock();
				}
					
				isBusyWith = null;
				if (msg == finalMsg) {
					success = true;
					break;
				}
			}
			while (mailboxST.isInQueue(finalMsg));
			
			busyLockFreed.signal();
			updateControllerQueue();
			return success;
		}
		finally {
			controller.getActorLock().unlock();
		}
	}
	
	/**
	 * Updates the registration of the ActorState in the controller.
	 * Should be invoked every time it may have changed before the actorLock is unlocked.
	 */
	public void updateControllerQueue() {
		int oldNumberOfThreadsNeeded = numberOfThreadsNeeded;
		numberOfThreadsNeeded = hasWorkForUnsynchronized();
		if (oldNumberOfThreadsNeeded == numberOfThreadsNeeded)
			return;
		controller.updateActorStateQueueUnsynchronized(this, oldNumberOfThreadsNeeded, numberOfThreadsNeeded);
	}
	
	/**
	 * Returns the Controller of the actor.
	 * @return the controller
	 */
	public Controller getController() {
		return controller;
	}

	/**
	 * Returns the actor number (used only for logging).
	 * @return the actorNumber
	 */
	public long getActorNumber() {
		return actorNumber;
	}

	/**
	 * Returns the state of the given actor instance.
	 * @param actor the actor
	 * @return the state
	 */
	public static ActorState getState(Actor actor) {
		ActorProxy ap = (ActorProxy) actor;
		return ap.getState__ACTORPROXYMETHOD();
	}
}
